package at.pimpmypipe.filter;

import java.io.StreamCorruptedException;
import java.security.InvalidParameterException;

import at.pimpmypipe.interfaces.Readable;
import at.pimpmypipe.interfaces.Writeable;

public abstract class AbstractSink<in> implements Writeable<in>, Runnable{

	private Readable<in> _input = null;

	public static Object ENDING_SIGNAL = null;
	
	public AbstractSink() {
		
	}

	public AbstractSink(Readable<in> input) throws InvalidParameterException {
		if (input == null) {
			throw new InvalidParameterException("input can't be null!");
		}
		_input = input;
	}

    protected in readInput() throws StreamCorruptedException{
        if (_input != null) {
            return _input.read();
        } else {
            throw new StreamCorruptedException("input is null");
        }
    }
	
	/**
	 * runs the filter in active-mode
	 */
	public void run() {
		in output = null;
		try {
			do {
				output = readInput();

				if (output != null) {
					write(output);
				}
			} while (output != null);

		} catch (StreamCorruptedException e) {
			System.out.print("Thread reports error: ");
			System.out.println(Thread.currentThread().getId() + " ("
					+ Thread.currentThread().getName() + ")");
			e.printStackTrace();
		}
	}

	public void setInput(Readable<in> input) {
		_input = input;
	}
}
